const log4js = require('log4js');
const logger = log4js.getLogger('log');
const uuid = require('uuid');
const fs = require('fs');
const {exec} = require('child_process');
const ipc = require('node-ipc');
const compressing = require('compressing');
const config = require('../config');
const { fstat } = require('fs');
ipc.config.id   = 'client';
ipc.config.retry= 100000;
ipc.config.logger = function(){};

const taskMap = new Map();
const timeoutDelay = 15000;
ipc.connectTo('crate', () => {
    console.log('connect to ipc crate server');
    ipc.of.crate.on(
        'finish_req',
        function(data){
            taskMap.set(data.taskId, data.status);
            ipc.log('disconnected from world'.notice);
        }
    );
});

const sleep = (time) => {
    return new Promise((resolve) => {
        setTimeout(() => {
            resolve();
        }, time);
    });
}

module.exports = {
    async '/:db/:table/_stream_load!POST' () {
        const {db, table} = this.params;
        const columns = this.ctx.headers['x-columns']

        // 保存数据入库
        // const ret = await this.DataProcess.saveBatch(this.params, {
        //     db,
        //     table,
        //     columns
        // });
        const thread = ipc.of['crate'];
        const taskId = uuid.v4();
        thread.emit('insert_data', {
            taskId,
            type: 'insert',
            sqlData: this.params,
            dbInfo: {
                db,
                table,
                columns
            }
        });
        const start = new Date().getTime();
        const ret = await new Promise(async (resolve) => {
            let taskStatus = taskMap.get(taskId);
            while (!taskStatus) {
                const end = new Date().getTime();
                // timeout
                if (end - start > timeoutDelay) {
                    resolve(-1);
                    break;
                }
                await sleep(200);
                taskStatus = taskMap.get(taskId);
            }
            if (taskStatus === 'finish') {
                resolve(1);
            } else {
                resolve(-1);
            }
        });
        const end = new Date().getTime();
        logger.info('ret', ret, 'time', (end - start) / 1000.0, 's');
        if (ret === -1) {
            this.ctx.status = 500;
            await this.json({
                code: 500,
            })
        } else {
            this.ctx.status = 200;
            await this.json({
                code: 200
            })
        }
    },

    async '/:db/:table/_stream_update!POST' () {
        const {db, table} = this.params;
        const columns = this.ctx.headers['x-columns'];
        const primaryField = this.ctx.headers['x-primary'];

        // 保存数据入库
        // const ret = await this.DataProcess.updateBatch(this.params, {
        //     db,
        //     table,
        //     columns,
        //     primaryField
        // });
        const thread = ipc.of['crate'];
        const taskId = uuid.v4();
        thread.emit('update_data', {
            taskId,
            type: 'update',
            sqlData: this.params,
            dbInfo: {
                db,
                table,
                columns,
                primaryField
            }
        });
        const start = new Date().getTime();
        const ret = await new Promise(async (resolve) => {
            let taskStatus = taskMap.get(taskId);
            while (!taskStatus) {
                const end = new Date().getTime();
                // timeout
                if (end - start > timeoutDelay) {
                    resolve(-1);
                    break;
                }
                await sleep(200);
                taskStatus = taskMap.get(taskId);
            }
            if (taskStatus === 'finish') {
                resolve(1);
            } else {
                resolve(-1);
            }
        });
        const end = new Date().getTime();
        logger.info('ret', ret, 'time', (end - start) / 1000.0, 's');
        if (ret === -1) {
            this.ctx.status = 500;
            this.json({
                code: 500
            })
        } else {
            this.ctx.status = 200;
            this.json({
                code: 200
            })
        }
    },

    async '/:db/:table/_file_load!ALL' () {
        const {db, table} = this.params;
        const file = this.ctx.request.files.file;
        if (file) {
            const filePath = file.path;
            const destPath = filePath.replace('.gz', '.csv');
            const ret = await new Promise((resolve) => {
                compressing.gzip.uncompress(filePath, destPath).then(() => {
                    const cmd = `${config.crashPath} -c "COPY ${db}.${table} FROM '${destPath}'"`;
                    exec(cmd, (error, stdout, stderr) => {
                        if (error) {
                            logger.error(`exec error: ${error}`);
                            resolve(`exec error: ${error}`);
                            return;
                        } else {
                            console.log(`stdout: ${file.name} ${stdout}`);
                            console.error(`stderr: ${file.name} ${stderr}`);
                            resolve(`${file.name} ${stdout} ${stderr}`);
                            fs.unlink(destPath, (err) => {
                                if (err) logger.error(err);
                                logger.info(destPath + ' was deleted');
                            });
                            fs.unlink(filePath, (err) => {
                                if (err) logger.error(err);
                                logger.info(filePath + ' was deleted');
                            });
                        }
                    });
                }).catch ((e) => {
                    logger.error(e);
                    resolve(e.toString());
                });
            });
            this.ctx.body = ret;
        }
    }
}